package com.atguigu.cdc;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

/**
 * @author: xu
 * @desc: 自定义序列化方式
 */
public class FlinkCdcCustomSchema {
    public static void main(String[] args) throws Exception {
        // TODO 1.准备流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        /*
        // TODO 2.开启检查点，FlinkCDC将读取binlog的位置信息以状态的方式保存在CK，想断点续传，需要从Checkpoint或者Savepoint启动程序
        // 2.1 开启Checkpoint,每隔5秒钟做一次CK  ,并指定CK的一致性语义
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        // 2.2 设置超时时间为1分钟
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        // 2.3 指定从CK自动重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, 2000L));
        // 2.4 设置任务关闭的时候保留最后一次CK数据
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 2.5 设置状态后端
        env.setStateBackend(new FsStateBackend("hdfs://node1:8020/flinkCDC"));
        // 2.6 设置访问HDFS的用户名
        // System.setProperty("HADOOP_USER_NAME", "root");
        */

        // TODO 3.创建Flink-MySQL-CDC的Source
        SourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname("node1")
                .port(3306)
                .username("root")
                .password("123456")
                .databaseList("gmall_realtime")
                // 可选配置项,如果不指定该参数,则会读取上一个配置中指定的数据库下的所有表的数据
                // 注意：指定的时候需要使用"db.table"的方式
                .tableList("gmall_realtime.t_user")
                .startupOptions(StartupOptions.initial())
                .deserializer(new MySchema())
                .build();

        // TODO 4.使用CDC Source从MySQL读取数据，打印输出
        env
                .addSource(sourceFunction)
                .print();

        // TODO 5.执行任务
        env.execute();
    }


    public static class MySchema implements DebeziumDeserializationSchema<String> {
        /*
          ConnectRecord {
              value=Struct {
                  after=Struct{id=1,name=liuxintong,age=17},
                  source=Struct {
                      db=gmall0224_realtime,
                      table=t_user
                  },
                  op=c
              },
          }
         */
        @Override
        public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
            // 获取变更的数据  value=Struct{after=Struct{id=3,name=ww11,age=55666}
            Struct valueStruct = (Struct) sourceRecord.value();
            Struct sourceStrut = valueStruct.getStruct("source");
            // 获取数据库的名称，表名
            String database = sourceStrut.getString("db");
            String table = sourceStrut.getString("table");

            // 获取操作类型 c->create
            String type = Envelope.operationFor(sourceRecord).toString().toLowerCase();
            if ("create".equals(type)) {
                type = "insert";
            }

            // TODO:获取影响的数据data
            //  源格式：id=1,name=liuxintong,age=17
            //  目标格式：{"id":74603,"order_id":28641,"order_status":"1005","operate_time":"2021-07-30 11:35:49"}}
            Struct afterStruct = valueStruct.getStruct("after");
            // 将变更数据封装为一个json对象
            JSONObject dataJsonObj = new JSONObject();
            if (afterStruct != null) {
                for (Field field : afterStruct.schema().fields()) {
                    Object o = afterStruct.get(field);
                    dataJsonObj.put(field.name(), o);
                }
            }

            // 创建JSON对象用于封装最终返回值数据信息
            JSONObject result = new JSONObject();
            result.put("database", database);
            result.put("table", table);
            result.put("type", type);
            result.put("data", dataJsonObj);

            // 发送数据至下游
            collector.collect(result.toJSONString());
        }

        @Override
        public TypeInformation<String> getProducedType() {
            return TypeInformation.of(String.class);
        }
    }
}
